package org.tea.infrastructure.ipc;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.BinaryJedisPubSub;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @author lxh
 */
public class CrawlRedisPubSub<T> extends BinaryJedisPubSub {

    private final static Logger log = LoggerFactory.getLogger(CrawlRedisPubSub.class);

    private final Lock lock;

    private final Condition condition;

    private final ConditionValueRef<T> valueRef;

    public CrawlRedisPubSub(Lock lock, Condition condition, ConditionValueRef<T> valueRef) {
        super();
        this.lock = lock;
        this.condition = condition;
        this.valueRef = valueRef;
    }

    @Override
    public void onMessage(byte[] channel, byte[] message) {
        super.onMessage(channel, message);
        lock.lock();
        try {
            String val = new String(message);
            if (log.isDebugEnabled()) {
                log.debug("receive message {} from {}", val, new String(channel));
            }
            valueRef.setValue(new ObjectMapper().readValue(val, new TypeReference<T>() {}));
            condition.signalAll();
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void onPMessage(byte[] pattern, byte[] channel, byte[] message) {
        super.onPMessage(pattern, channel, message);
        lock.lock();
        try {
            String val = new String(message);
            if (log.isDebugEnabled()) {
                log.debug("receive pattern {} message {} from {}",
                        new String(pattern), val, new String(channel)
                );
            }
            valueRef.setValue(new ObjectMapper().readValue(val, new TypeReference<T>() {}));
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        } finally {
            condition.signalAll();
            lock.unlock();
        }
    }
}
